package org.ws.task.netty.support;

import com.google.common.util.concurrent.Uninterruptibles;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.model.protocol.base.RequestMessage;
import org.ws.task.model.protocol.base.ResponseMessage;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;

public class ResponseFuture {

    private static Logger log = LoggerFactory.getLogger(ResponseFuture.class.getName());

    private final AtomicReference<Semaphore> semaphoreAtomic;
    private final Consumer<ResponseFuture> callback;
    private final CountDownLatch waitResponseLatch;

    private final Channel channel;

    private final RequestMessage request;

    private final long timeout;

    private ResponseMessage response;

    private Throwable cause;

    private volatile boolean sendSuccess;

    public void setResponse(ResponseMessage response) {
        this.response = response;
    }

    public void setCause(Throwable cause) {
        this.cause = cause;
    }

    public void setSendSuccess(boolean sendSuccess) {
        this.sendSuccess = sendSuccess;
    }

    public Channel getChannel() {
        return channel;
    }

    public RequestMessage getRequest() {
        return request;
    }

    public long getTimeout() {
        return timeout;
    }

    public ResponseMessage getResponse() {
        return response;
    }

    public Throwable getCause() {
        return cause;
    }

    public boolean isSendSuccess() {
        return sendSuccess;
    }

    public ResponseFuture(Channel channel, RequestMessage request, long timeout, Semaphore semaphore, Consumer<ResponseFuture> callback) {
        this.channel = channel;
        this.request = request;
        this.timeout = timeout;
        this.semaphoreAtomic = new AtomicReference<>(semaphore);
        this.callback = callback;
        this.waitResponseLatch = new CountDownLatch(1);
    }

    public void release(){
        if(semaphoreAtomic.get() != null) {
            semaphoreAtomic.getAndSet(null).release();
        }
    }

    public boolean syncAwait() {
        return Uninterruptibles.awaitUninterruptibly(this.waitResponseLatch,this.timeout,TimeUnit.MILLISECONDS);
    }

    protected void executeCallback(final ExecutorService executorService){
        if (this.waitResponseLatch.getCount()>0) {
            this.waitResponseLatch.countDown();
            if(callback != null){
                executorService.submit(()->{
                    try {
                        callback.accept(ResponseFuture.this);
                    } catch (Exception e) {
                        log.error(e.getMessage(),e);
                    }
                    finally {
                        ResponseFuture.this.release();
                    }
                });
            }
            else {
                this.release();
            }
        }
    }

}
